package com.motu.data;

import com.motu.vertx.module.utility.kafka.KafkaManager;
import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.client.WebClientOptions;
import io.vertx.kafka.client.producer.KafkaProducer;
import io.vertx.kafka.client.producer.KafkaProducerRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * 魔兔BI数据管理父类
 */
public class MotuBiBaseManager {

	private static final Logger logger = LoggerFactory.getLogger("MotuBiManager");
	public static KafkaProducer<String, String> bi_producer;
	public static WebClient webClient;
	public static final String KAFKA_TOPIC_BI = "log_colloction_bi";

	/**
	 * 初始化魔兔BIkafka配置（需要在进程初始化调用）
	 *
	 * @param vertx
	 * @param appConfig
	 */
	public static void init(Vertx vertx, JsonObject appConfig) {
		webClient = WebClient.create(vertx, new WebClientOptions().setKeepAlive(true));

		// 数据中台 游戏行为事件 kafka地址
		String address = appConfig.getString("bi-collection-kafka-address");
		if (!Tool.isEmpty(address)) {
			Properties bi_config = new Properties();
			bi_config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, address);
			bi_config.put("key.serializer", StringSerializer.class);
			bi_config.put("value.serializer", StringSerializer.class);
			bi_config.put("acks", "1");
			// 优化kafka生产者配置
			KafkaManager.upgradeProducerConfig(bi_config, appConfig);
			// 创建一个Kafka Producer
			bi_producer = KafkaProducer.create(vertx, bi_config);
			logger.info("motu bi kafka init success address:{}", address);
		}
	}

	/**
	 * 日志发送到消息队列
	 *
	 * @param params 发送参数
	 */
	public static void pushToQueue(JsonObject params) {
		logger.info("MotuBiManager send params:" + params);
		String topicName = KAFKA_TOPIC_BI;
		if (bi_producer == null) {
			logger.error("MotuBiManager pushDBQueue fail producer is null!! topicName:" + topicName);
			return;
		}
		KafkaProducerRecord<String, String> record = KafkaProducerRecord.create(topicName, params.toString());
		bi_producer.send(record, done -> {
			if (done.failed()) {
				logger.error("MotuBiManager pushDBQueue fail params:{} cause:{}", params, Tool.getException(done.cause()));
			} else {
//				RecordMetadata recordMetadata = done.result();
			}
		});
	}

}
